package com.mall.order.rocketmq;

import com.mall.order.OrderQueryService;
import com.mall.order.constants.OrderConstants;
import com.mall.order.dal.entitys.Order;
import com.mall.order.dal.entitys.Stock;
import com.mall.order.dal.persistence.OrderMapper;
import com.mall.order.dal.persistence.StockMapper;
import com.mall.order.dto.OrderDetailRequest;
import com.mall.order.dto.OrderDetailResponse;
import com.mall.order.dto.OrderItemDto;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.Reference;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * @author xudengfeng
 * @date 2020/12/23 21:46
 */
@Slf4j
@Component
public class CommonOrderMqConsumer {


    @Autowired
    OrderMapper orderMapper;

    @Autowired
    StockMapper stockMapper;

    @Reference(check = false)
    OrderQueryService orderQueryService;

    private DefaultMQPushConsumer mqPushConsumer;

    @PostConstruct
    public void init() {
        mqPushConsumer = new DefaultMQPushConsumer("consumer_first");

        mqPushConsumer.setNamesrvAddr("localhost:9876");
        try {
            mqPushConsumer.subscribe("order_topic", "*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        mqPushConsumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                try {
                    MessageExt messageExt = list.get(0);
                    byte[] body = messageExt.getBody();
                    String orderId = new String(body);

                    // 根据orderId查询订单
                    Order order = orderMapper.selectByPrimaryKey(orderId);
                    Integer status = order.getStatus();

                    if (status == OrderConstants.ORDER_STATUS_PAYED) {
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }

                    if (status == OrderConstants.ORDER_STATUS_INIT) {
                        // 修改订单状态  回补库存
                        Order orderUpdate = new Order();
                        orderUpdate.setOrderId(orderId);
                        orderUpdate.setStatus(OrderConstants.ORDER_STATUS_TRANSACTION_CANCEL);
                        orderMapper.updateByPrimaryKeySelective(orderUpdate);
                        // stockMapper.updateStock();
                        // 通过订单id查找商品,并将商品的库存补回去

                        OrderDetailRequest request = new OrderDetailRequest();
                        request.setOrderId(orderId);
                        OrderDetailResponse response = orderQueryService.queryOrdersById(request);
                        List<OrderItemDto> goodsList = response.getGoodsList();
                        for (OrderItemDto orderItemDto : goodsList) {
                            // 订单中的商品Id
                            String itemId = orderItemDto.getItemId();
                            // 订单中的商品的数量
                            Integer num = orderItemDto.getNum();

                            Stock stock = new Stock();
                            stock.setItemId(Long.valueOf(itemId));

                            // 先查出库存,再补库存

                            Stock stockRet = stockMapper.selectByPrimaryKey(stock);
                            Long stockCount = stockRet.getStockCount();
                            stockRet.setStockCount(stockCount + num);

                            int rows = stockMapper.updateByPrimaryKeySelective(stockRet);
                            if (rows == 1) {
                                log.info("取消订单并修改库存成功>>><<<");
                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                            }
                        }
                        // 回补库存以后才返回成功,否则重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });

        try {
            mqPushConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

}
